#include "global_vars.h"
double latency[1000000];

//------------------------------------------------------------------------------
double get_mili_seconds() {
	struct timeval mtv;
	gettimeofday(&mtv, NULL);
	double time = 1000 * mtv.tv_sec + mtv.tv_usec / (double) 1000;
	return time;
}

int get_predecessot(int p_id) {

	int predecessor;
	if (p_id == 0)
		predecessor = get_config_info()->num_nodes - 1;
	else
		predecessor = p_id - 1;
	return predecessor;
}

int get_pre_predecessor(int p_id) {
	int pre_predecessor;
	int predecessor = get_predecessot(p_id);

	if (predecessor == 0)
		pre_predecessor = get_config_info()->num_nodes - 1;
	else
		pre_predecessor = predecessor - 1;
	return pre_predecessor;
}

//------------------------------------------------------------------------------
void utoBroadcast(int fd, short event, void * arg) {
	int i, interval_msg_sent = 0;
	msg *vp;
	int self_id = get_config_info()->self_id;
	while (interval_msg_sent < submit_per_wake) {
		vector_clock[self_id]++;
		msg *new_msg = (msg*) malloc(sizeof(msg));
		new_msg->type = 'm';
		new_msg->p_id = self_id;

		for (i = 0; i < get_config_info()->num_nodes; i++)
			new_msg->vC[i] = vector_clock[i];

		new_msg->status = false;
		new_msg->startMiliSeconds = get_mili_seconds();
		vp = (msg*) (rbsearch((void *) new_msg, myTree));
		assert(vp != NULL);
		int status = send(get_config_info()->nodes[self_id].successor_sid, new_msg, sizeof(msg), MSG_WAITALL);
		assert(status == sizeof(msg));
		interval_performance->interval_bytes_proposed += sizeof(msg);
		interval_performance->interval_num_packets_propsoed++;
		interval_msg_sent++;
	}
	evtimer_add(timeout1, &tv1);
}
//------------------------------------------------------------------------------
void tryDeliver() {
	int self_id = get_config_info()->self_id;
	void *vp;
	head = (msg*) (rbmin(myTree));// I think even if you remove this, this will not change anything, or the one at the end could be removed?
	//	head = rblookup(RB_LUFIRST, NULL, myTree);

	while (head != NULL && head->status) {
		//		if (head->p_id == 4)
		{ // It is strange that it is working like this.this guy has the highest delivery rate! the one in the middle
			double lat = get_mili_seconds() - head->startMiliSeconds;
			interval_performance->num_packets_delivered_latency++;
			interval_performance->interval_latency += lat;

			if (performance_index > DISCARD && performance_index < STOPPING_CONDITION - DISCARD) {
				latency[interval_performance->latency_counter] = lat;
				interval_performance->latency_counter++;

			}
		}
		//		int i;
		//
		//		for (i = 0; i <= 4; i++)
		//			printf("%d", (head)->vC[i]);
		//
		//		printf("\n");

		interval_performance->interval_num_packets_delivered++;
		vp = (msg*) rbdelete(head, myTree);
		assert(vp != NULL);
		head = (msg*) (rbmin(myTree));
		//		head =  rblookup(RB_LUFIRST, NULL, myTree);
		free(vp);

	}
}
//------------------------------------------------------------------------------
void message_processor(int fd, short event, void * arg) {
	int status;
	char t;

	status = recv(fd, &t, sizeof(char), MSG_PEEK);
	assert(status == sizeof(char));
	interval_performance->num_interrupts++;
	switch (t) {
	case 'm':
		handel_proposal_msg(fd);
		break;
	case 'a':
		handel_ack_msg(fd);
		break;
	default: {
		printf("Unexpected message type received.%c\n", t);
	}
	}

}

void handel_proposal_msg(int fd) {
	int i, status, predecessor = 0, pre_predecessor = 0;
	msg *vp;

	msg recv_msg;
	ack recv_ack;
	int self_id = get_config_info()->self_id;

	status = recv(fd, &recv_msg, sizeof(msg), MSG_WAITALL);
	assert(status == sizeof(msg));
	interval_performance->interval_bytes_recv += status;
	predecessor = get_predecessot(recv_msg.p_id);
	pre_predecessor = get_pre_predecessor(recv_msg.p_id);

	if ((recv_msg.vC[recv_msg.p_id]) > (vector_clock[recv_msg.p_id])) {
		if (get_config_info()->self_id != predecessor) {

			status = send(get_config_info()->nodes[self_id].successor_sid, &recv_msg, sizeof(msg), MSG_WAITALL);
			assert(status == sizeof(msg));
			interval_performance->interval_bytes_sent += status;
			msg *new_msg = (msg*) malloc(sizeof(msg));
			memcpy(new_msg, &recv_msg, sizeof(msg));

			vp = (msg*) (rbsearch((void *) new_msg, myTree));
			assert(vp != NULL);

		} else {

			//			assert(get_config_info()->self_id == predecessor);

			msg *new_msg = (msg*) malloc(sizeof(msg));
			memcpy(new_msg, &recv_msg, sizeof(msg));
			new_msg->status = true;

			vp = (msg*) (rbsearch((void *) new_msg, myTree));
			assert(vp != NULL);
			//			assert(vp->status == true);

			recv_ack.type = 'a';
			recv_ack.p_id = recv_msg.p_id;
			for (i = 0; i < get_config_info()->num_nodes; i++)
				recv_ack.vC[i] = recv_msg.vC[i];
			recv_ack.status = false;

			tryDeliver();
			//			sleep(10);
			status = send(get_config_info()->nodes[self_id].successor_sid, &recv_ack, sizeof(ack), MSG_WAITALL);
			assert(status == sizeof(ack));
			interval_performance->interval_bytes_sent += status;

		}
		updateClock(vector_clock, recv_msg.vC);

	} else
		printf("OOPS! id %d: (%d < %d)\n", (recv_msg.p_id), (recv_msg.vC[recv_msg.p_id]), (vector_clock[recv_msg.p_id]));

}
void handel_ack_msg(int fd) {
	int i, status, predecessor = 0, pre_predecessor = 0;
	msg *vp;
	int self_id = get_config_info()->self_id;
	msg recv_msg;
	ack recv_ack;

	status = recv(fd, &recv_ack, sizeof(ack), MSG_WAITALL);
	assert(status == sizeof(ack));

	interval_performance->interval_bytes_recv += status;
	predecessor = get_predecessot(recv_ack.p_id);
	pre_predecessor = get_pre_predecessor(recv_ack.p_id);

	recv_msg.p_id = recv_ack.p_id;
	recv_msg.type = 'm';

	for (i = 0; i < get_config_info()->num_nodes; i++)
		recv_msg.vC[i] = recv_ack.vC[i];

	recv_msg.status = recv_ack.status;

	vp = (msg*) (rbfind(&recv_msg, myTree));
	assert(vp != NULL);
	vp->status = true;
	tryDeliver();
	if (get_config_info()->self_id != pre_predecessor) {
		//		sleep(10);
		status = send(get_config_info()->nodes[self_id].successor_sid, &recv_ack, sizeof(ack), MSG_WAITALL);
		assert(status == sizeof(ack));
		interval_performance->interval_bytes_sent += status;
	}

}
